#include "util/serialize.hxx"
#include "util/fs.hxx"
#include "util/io.hxx"
#include "util/common.hxx"
#include "log/logmanager.hxx"
#include "log/logfile.hxx"
#include "dbenv.hxx"
#include "db/dbenv_impl.hxx"
#include "util/lockguard.hxx"
#include "db/dbpartition.hxx"
#include "db/datastore_impl.hxx"
#include "db/addressformat.hxx"
#include "util/logger.hxx"
#include "db/metafile.hxx"
#include <algorithm>
#include <assert.h>


LogManager::LogManager(DbEnvImpl& env, uint32_t flag)
: m_activeLogFile(NULL),
  m_capacity(10*1024*1024),
  m_curLogInBuffer(0), 
  m_env(&env), 
  m_nextScn(0),
  m_inRecovery(false),
  m_flag(flag) {

}


LogManager::~LogManager() {

	if(m_activeLogFile) {delete m_activeLogFile; m_activeLogFile = NULL;}
}

DbStatus LogManager::init() {

	DbStatus ret;

    pthread_mutex_init(&m_mutex, NULL);
    pthread_mutex_init(&m_mutexList, NULL);

	m_bufferSize = m_env->getLogBufferSize();
	
	
	//get last ID, if not exists create next new one
	ret = loadLogFiles();
	if(!ret.isok()) return ret; 

	if(m_logfiles.empty()) {
	
		LogFileWriter * lfw;
		ret = nextLogFile(lfw);
		if(!ret.isok()) return ret; 

        m_activeLogFile = lfw;

        LockGuard lockg(m_mutexList);
        m_lfwList.push_back(m_activeLogFile);

	}
	else {
	
		uint32_t lastLogId = getLastLogId();
		m_activeLogFile = new LogFileWriter(lastLogId, m_env->getLogpath(), m_capacity);
		ret = m_activeLogFile->init(); if(!ret.isok()) return ret; 

		if(m_activeLogFile->isSynced() ) {
		
			delete m_activeLogFile; m_activeLogFile = NULL;
			
			ret = nextLogFile(m_activeLogFile); 
			if(!ret.isok()) return ret;

            LockGuard lockg(m_mutexList);
            m_lfwList.push_back(m_activeLogFile);
        }

        LockGuard lockg(m_mutexList);
        m_lfwList.push_back(m_activeLogFile);
	}

    //do recovery
    ret = doRecovery();
    if(!ret.isok()) return ret;

    Logger::log("initialize redo log successfully, flag: %d", m_flag);

	return ret;
}

DbStatus LogManager::flush() {

    DbStatus ret;

    LockGuard lock(m_mutex);

    if(!m_inRecovery) ret = m_activeLogFile->flush();
    m_curLogInBuffer = 0;

	return ret;
}

std::set<uint32_t> LogManager::getLogsTobesynced() {

    LockGuard lockg(m_mutexList);

    std::set<uint32_t> v;

    //return all current log files not synced except for the the active log file
    for(std::list<LogFileWriter*>::iterator iter = m_lfwList.begin(); iter != m_lfwList.end(); ++iter) {

        if((*iter)->getLogId() != m_activeLogFile->getLogId()) {
        
            v.insert((*iter)->getLogId());
        } 
    }

    return v;
}


DbStatus LogManager::writeSyncFlag(std::set<uint32_t>& logv) {

	DbStatus ret;

    LockGuard lockg(m_mutexList);

	//update the m_synced of the log file header in logv list
    for(std::list<LogFileWriter*>::iterator iter = m_lfwList.begin(); iter != m_lfwList.end(); ) {
    
        if( logv.count( (*iter)->getLogId() ) > 0 ) {
        
            LogFileWriter * lfw = *iter;
            lfw->setSynced();

            delete lfw;
            iter = m_lfwList.erase(iter);
        }
        else ++iter;
    }


	return ret;
}


DbStatus LogManager::addRecord(LogRecord& rec) {

	DbStatus ret;

    LockGuard lock(m_mutex);


	rec.setScn(nextScn());

    //the buffer dbt is being reused, so doesn't need to free 
    Dbt* dbt = rec.serialize();
	

	ret = m_activeLogFile->append(dbt); 
	if(ret.getCode() == DbStatus::DB_LOGOVERFLOW) {
	
		ret = m_activeLogFile->flush(); if(!ret.isok()) return ret;
        m_activeLogFile->closeWriter();
        m_curLogInBuffer = 0;

		LogFileWriter * lf = NULL;
		ret = nextLogFile(lf);  if(!ret.isok()) return ret;
       
        m_activeLogFile = lf;

        ret = m_activeLogFile->append(dbt);

        LockGuard lockg(m_mutexList);
        m_lfwList.push_back(m_activeLogFile);
	}


    m_curLogInBuffer += dbt->getDatalen();

	if( (m_flag & LOG_SYNC) || m_curLogInBuffer > m_bufferSize) {
   
        ret = m_activeLogFile->flush(); if(!ret.isok()) return ret;
        m_curLogInBuffer = 0;
    }


	return ret;
}


DbStatus LogManager::doRecovery() {

	DbStatus ret; 

    m_inRecovery = true;

	for(std::list<std::string>::iterator iter = m_logfiles.begin(); iter !=
            m_logfiles.end(); ++iter) {

		ret = recovery(atoi(iter->c_str()), m_env->getLogpath()); 
        if(!ret.isok()) { 

            m_inRecovery = false;
            return ret;
        }
	}

    m_inRecovery = false;

	return ret;
}


DbStatus LogManager::recovery(uint32_t logId, const std::string& logPath) {

	DbStatus ret;

	LogFileReader reader(logId, logPath);

	ret = reader.init(); if(!ret.isok()) return ret; 
	if(reader.isSynced()) return ret;

	std::vector<LogRecord> records;
	ret = reader.getAllRecords(records);
	if(records.empty() || (!ret.isok() && DbStatus::DB_CORRUPTION != ret.getCode()) ) return ret;
    

	for(int i = 0; i < records.size(); ++i) {
	
		if(!canbeApplied(records[i])) continue;
		ret = apply(records[i]); if(!ret.isok() && DbStatus::DB_SEGMENTNOTFOUND != ret.getCode()) return ret;
	    records[i].clear();
    }

    //sync the changes from the log
    ret = m_env->sync();
    if(!ret.isok()) return ret;

    //change sync flag in the meta file, except for the last logfile
    if(logId != getLastLogId()) reader.setSynced();

	return ret;
}


uint32_t LogManager::getLastLogId() {

    uint32_t id = 0;
	if(!m_logfiles.empty()) id = atoi(m_logfiles.rbegin()->c_str());

	return id;
}


DbStatus LogManager::loadLogFiles() {

	DbStatus ret;

	std::vector<std::string> children;
	ret = FsUtility::getChildren(m_env->getLogpath(), children);
	if(!ret.isok()) return ret; 
	
    m_logfiles.clear();

	for(int i = 0; i < children.size(); ++i) {
	
 		if(isvalidLogFileName(children[i]) &&
                !FsUtility::isDir(m_env->getLogpath() + "/" + children[i])) {
		
			m_logfiles.push_back(children[i].substr(0, children[i].size() - 4));
		}
	}

    m_logfiles.sort(LogFileNameComp());

	return ret;
}

bool LogManager::isvalidLogFileName(const std::string& logName) {

	int i;
	for(i = 0; i < logName.size(); ++i) {
	
		if(!isdigit(logName[i])) break;
	}

	if(i + 4 != logName.size() ) return false;
	
	return 0 == strncmp(&logName[i], ".log", 4);
}


std::string LogManager::getLogFilePath(const std::string& logName) {

    return  m_env->getLogpath() + "/" + logName;
}

std::string LogManager::getLogFilePath(uint16_t logID) {

    return  m_env->getLogpath() + "/" + utility::itoa(logID) + ".log";
}


uint64_t  LogManager::nextScn() {

	m_nextScn++;

	return m_nextScn;
}



DbStatus  LogManager::nextLogFile(LogFileWriter* & lf) {

	DbStatus ret;

	uint32_t logId = 0;
	if(m_activeLogFile) logId = m_activeLogFile->getLogId();

	logId++;
	
	//creae a new log file
	lf = new LogFileWriter(logId, m_env->getLogpath(), m_capacity);
	ret = lf->init(); if(!ret.isok()) return ret; 

    MetaFile::getInstance()->setLogSync(logId, 0);

    m_logfiles.push_back(utility::itoa(logId));
    if(m_logfiles.size() > 100) {
   
        std::string logfile =  getLogFilePath(m_logfiles.front() + ".log"); 
        ret = FsUtility::deleteFile(logfile);
        m_logfiles.pop_front();
    }


	return ret;
}

bool LogManager::canbeApplied(LogRecord& rec) {

    //map partition name to its lwn and hwm
    //we store its temporary lwm and hwm in the map below
    static std::map<std::string, uint64_t> tlwm, thwm;

    uint64_t scn = rec.getScn();
    std::string dsName = rec.getDsName();
    std::string partitionName = rec.getPartitionName();

    uint64_t lwm, hwm;
    if(tlwm.count(partitionName) > 0) {

        lwm = tlwm[partitionName];
        hwm = thwm[partitionName];
    }
    else {

        DataStoreImpl* store = m_env->findStore(dsName);
        DbPartition* partition = store->findPartition(partitionName);

        partition->getWaterMarks(lwm, hwm);
        tlwm[partitionName] = lwm;
        thwm[partitionName] = hwm;
    }

    bool ret = false;

    if(lwm == hwm) {
    
        //index array is consistent
        ret = (scn > lwm);
        if(ret) {

            tlwm[partitionName] = scn;
            thwm[partitionName] = scn;
        }
    }
    else if(lwm < hwm) {
    
        //not consistent
        ret = (scn > lwm && scn <= hwm);
        if(ret) tlwm[partitionName] = scn;
    }
    else {
    
        //index array is corrupted which can be detected by indexarray, so should not reach here
    }

    return ret;
}


DbStatus LogManager::apply(LogRecord& rec) {

	DbStatus ret;

    uint32_t type = rec.getType();

    std::string dsName = rec.getDsName();
    std::string partitionName = rec.getPartitionName();

    DataStoreImpl* store = m_env->findStore(dsName);
    DbPartition* partition = store->findPartition(partitionName);

    uint64_t newAddr = rec.getAddrNew();
    uint64_t oldAddr = rec.getAddrOld();

    AddressFormat naf(newAddr), oaf(oldAddr);
    uint16_t newSegID = naf.getSegID();
    uint32_t newOffset = naf.getOffset();
    uint32_t newDataSize = naf.getDataSize();
    uint16_t oldSegID = oaf.getSegID();
    uint32_t oldOffset = oaf.getOffset();
    uint32_t oldDataSize = oaf.getDataSize();


    //update new segment: data,  loadsize, and indexarray
	uint32_t loadSizeNew = rec.getLoadSizeNew();
	std::string data = rec.getDataNew();
	uint32_t index = rec.getIndex();
    uint64_t scn = rec.getScn();

    if(scn > m_nextScn) m_nextScn = scn;

    //if newAddr==0, it indicates that it is log record corresponding a deletion
    //and only 1 value in the bucket of the key
    if(newAddr) {

        if(type == LogRecord::COMPACT) {
        
            //copy data from old segment to new segment
            std::string oldData;
            ret = partition->get(oldSegID, oldOffset, oldDataSize, oldData);  if(!ret.isok()) return ret;
            data = oldData;

        }

        ret = partition->put(newSegID, newOffset, data, loadSizeNew, index, newAddr, scn);

        //compact may already reclaimed the segment, we can continue recovery in this case
        //this can happen when we are doing catastrophe recovery
        if(!ret.isok() &&  DbStatus::DB_SEGMENTNOTFOUND != ret.getCode()) return ret;
    }

	//update loadsize of old segment
	uint32_t loadSizeOld = rec.getLoadSizeOld();
	if(oldSegID > 0) ret = partition->setLoadSize(oldSegID, loadSizeOld);

	return ret;
}

void LogManager::switchLog() {
}

